package com.event.ingestion.common;

import com.event.ingestion.config.LoadConfig;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.Mongo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.security.acl.LastOwnerException;
import java.util.Properties;

public class MongoWriter implements Persistable {
    private String mongoHost = "127.0.0.1";
    private int mongoPort = 27017;
    private String mongoDatabase = "events_db";
    private String mongoUser = "it21";
    private String mongoPwd = "password";
    private String mongoCollection ;
    private Parsable<Tuple<BasicDBObject,BasicDBObject>> parser;
    public MongoWriter(String mongoCollection,Parsable<Tuple<BasicDBObject,BasicDBObject>> parser){
        this.mongoCollection=mongoCollection;
        this.parser=parser;

    }




    @Override
    public void initialize(Properties properties) {
      this.mongoHost=properties.getProperty(LoadConfig.mongoHost);
      try {
          this.mongoPort=Integer.parseInt(properties.getProperty(LoadConfig.mongoPort));

      }catch (Exception e){
          this.mongoPort=27017;
      }
      this.mongoDatabase=properties.getProperty(LoadConfig.mongoDatabase);
      this.mongoUser=properties.getProperty(LoadConfig.mongoUser);
      this.mongoPwd=properties.getProperty(LoadConfig.mongoPwd);
    }

    @Override
    public int write(ConsumerRecords<String, String> records) throws Exception {
        int numDocs=0;
        //the mongo-client
        Mongo  mongo=new Mongo(mongoHost,mongoPort);
       try {
           DB db = mongo.getDB(mongoDatabase);
           if (mongoUser!=null && !mongoUser.isEmpty()){
               //authenticate
           db.authenticate(mongoUser,mongoPwd.toCharArray());
           }
           //get the collection
           DBCollection collection = db.getCollection(mongoCollection);
           //flags
           long passHead=0;
           for (ConsumerRecord<String,String> record:records) {
               try {
                   String[] elements = record.value().split(",", -1);
                    if (passHead==0 && this.parser.isHeader(elements)){
                        passHead=1;
                        continue;
                    }
                    if (this.parser.isValid(elements)){
                        Tuple<BasicDBObject, BasicDBObject> kv = this.parser.parse(elements);
                        collection.update(kv.key,new BasicDBObject("$set",kv.value),true,false);
                        numDocs++;
                    }
                    else {
                        System.out.println(String.format("ErrorOccured: invalid message found when writing to MongoDB! - %s",record.value()));
                    }
               }catch (Exception e){
                   System.out.println("ErrorOccured: " + e.getMessage());
               }
           }


       }finally {
           mongo.close();
       }


        return numDocs;
    }
}
